package com.asiainfo.dacp.datastash.broker;

import com.asiainfo.dacp.core.BeanFactory;
import com.asiainfo.dacp.datastash.broker.core.BrokerRegistryService;
import com.asiainfo.dacp.datastash.broker.facade.service.UnitInitService;
import com.asiainfo.dacp.datastash.broker.leader.LeaderService;
import com.asiainfo.dacp.datastash.broker.listener.BrokerListenerManager;
import com.asiainfo.dacp.datastash.broker.schedule.StashSchedulerInternal;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLoggerService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.InputStreamReader;

/**
 * Broker服务的启动类
 *
 * 本服务类是无状态的，所有的状态都在redis中进行保存，所有的操作都使用共享内存完成。
 * 可以启动多个实例完成负载的能力，启动该类时会首先检查redis是否存在，不存在不能完成启动，
 * redis出现问题整个服务挂起，并进行严重警告
 *
 * 1.启动调度进程的实例调度能力，触发任务后交由dispatcher进行任务分派
 * 2.启动健康监控进程，完成对于broker的识别
 * 3.启动http服务完成外部能力接口的启动
 *
 * @author MeiKefu
 * @date 2017-10-31
 */
@Component
public class StashBrokerBoot {

    private static Logger LOG = LoggerFactory.getLogger(StashBrokerBoot.class);

    @Autowired
    private StashTraceLogger tracelog;
    
    @Autowired
    private UnitInitService stashUnitInitService;
    
    @Autowired
    private LeaderService leaderService;
    
    private String clzName = this.getClass().getSimpleName();


    public void init(){
    	
    	LOG.info("注册Broker信息");
    	BrokerRegistryService service = BeanFactory.getBean(BrokerRegistryService.class);
    	service.registry();
    	
    	/**
    	 * TODO 初始化内存，后续考虑是否放在redis中
    	 * 
    	 * */
    	stashUnitInitService.initUnit();	
    	stashUnitInitService.updateUnitBatchNo();

        LOG.info("启动内置调度服务");
        //1.启动调度
        /**
         * TODO 增量更新内存信息
         */
        StashSchedulerInternal server = BeanFactory.getBean(StashSchedulerInternal.class);
        server.init();
        tracelog.info(new StashTraceLogVo(clzName,"broker:boot","启动内置调度服务"));
        
        /**
         * 启动选主
         */
        LOG.info("启动选主");
        leaderService.start();
        
        
        LOG.info("启动内置监听服务");
        //2.启动监听器
        BrokerListenerManager queueListenerManager = new BrokerListenerManager();
        queueListenerManager.startup();
        tracelog.info(new StashTraceLogVo(clzName, "broker:boot","启动内置监听服务"));

        StashTraceLoggerService stashTraceLoggerService = BeanFactory.getBean(StashTraceLoggerService.class);
        //stashTraceLoggerService.createCdrTable();

    }

    public static void main(String[] args){
        StashBrokerBoot boot =null;
        try {

            //TODO 日志的问题，启动的常驻是否合理？

            LOG.info("DataStash-Broker启动");

            ApplicationContext context = new FileSystemXmlApplicationContext(new String[]{"classpath:conf/spring/*.xml","conf/spring/*.xml"});
            boot = context.getBean(StashBrokerBoot.class);

            boot.init();

            String br = new BufferedReader(new InputStreamReader(System.in)).readLine();

            if("quit".equalsIgnoreCase(br)){
                LOG.info("DataStash-Broker中止");
                System.exit(0);
            }

        } catch (Throwable ex) {
            ex.printStackTrace();
            System.exit(1);
        }
    }
}
